/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <stdio.h>
#include <stdlib.h>
#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);

const static int KEY_CONDITION = 1;
const static int VALUE_CONDITION = 2;

bool keyConditionExists(const vector<SelCond> cond);
bool valueConditionExists(const vector<SelCond> cond);
bool sortTuplesByPid(const pair<int, RecordId> p1, const pair<int, RecordId> p2);
bool operator<(const pair<int, RecordId> p1, const pair<int, RecordId> p2);

RC findMatchingTuples(BTreeIndex index, const vector<SelCond> cond,
                      vector< pair<int, RecordId> >& matchingTuples);
RC checkValueConditionsAndPrint(vector< pair <int, RecordId> > matchingTuples,
                                const vector<SelCond> cond, RecordFile rf,
                                int attr, const string table);

bool operator<(const pair<int, RecordId> p1, const pair<int, RecordId> p2)
{
  return p1.first <  p2.first;
}

bool sortTuplesByPid(const pair<int, RecordId> p1, const pair<int, RecordId> p2)
{
  return p1.second.pid <  p2.second.pid;
}

RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

RC findMatchingTuples(BTreeIndex index, const vector<SelCond> cond,
                      vector< pair<int, RecordId> >& matchingTuples)
{
  RC     rc;
  int    key;
  string value;
  int    diff;
  RecordId rid;

  for (int i = 0; i < cond.size(); ++i)
  {
    if (cond[i].attr == KEY_CONDITION && cond[i].comp != SelCond::NE)
    {
      int searchKey = atoi(cond[i].value);
      IndexCursor cursor;

      vector <pair<int, RecordId> > matchingTuplesForCondition;
      switch (cond[i].comp)
      {
        case SelCond::EQ:
          index.locate(searchKey, cursor);
          index.readForward(cursor, key, rid);
          if (key == searchKey)
          {
            pair<int, RecordId> tuple;
            tuple.first = key;
            tuple.second = rid;
            matchingTuplesForCondition.push_back(tuple);
          }
          break;
        case SelCond::GE:
          searchKey -= 1;
        case SelCond::GT:
          index.locate(searchKey, cursor);
          while ((rc = index.readForward(cursor, key, rid)) != RC_END_OF_TREE)
          {
            if (key > searchKey)
            {
              pair<int, RecordId> tuple;
              tuple.first = key;
              tuple.second = rid;
              matchingTuplesForCondition.push_back(tuple);
            }
          }
          break;
        case SelCond::LE:
          searchKey += 1;
        case SelCond::LT:
          index.locate(index.minKey, cursor);
          while ((rc = index.readForward(cursor, key, rid)) != RC_END_OF_TREE)
          {
            if (key < searchKey)
            {
              pair<int, RecordId> tuple;
              tuple.first = key;
              tuple.second = rid;
              matchingTuplesForCondition.push_back(tuple);
            }
            else
            {
              break;
            }
          }
          break;
        default:
          break;
      }

      if (matchingTuplesForCondition.empty())
      {
        matchingTuples.clear();
        break;
      }

      if (i == 0)
      {
        matchingTuples.insert(matchingTuples.begin(), matchingTuplesForCondition.begin(),
                              matchingTuplesForCondition.end());
      }
      else
      {
        sort(matchingTuples.begin(), matchingTuples.end());
        sort(matchingTuplesForCondition.begin(), matchingTuplesForCondition.end());
        vector< pair<int, RecordId> > intersectingTuples;
        set_intersection(matchingTuples.begin(), matchingTuples.end(),
                         matchingTuplesForCondition.begin(), matchingTuplesForCondition.end(),
                         back_inserter(intersectingTuples));
        matchingTuples.clear();
        matchingTuples.insert(matchingTuples.begin(), intersectingTuples.begin(), intersectingTuples.end());
      }

      if (matchingTuples.empty())
      {
        break;
      }
    }
  }
  return 0;
}

RC checkValueConditionsAndPrint(vector< pair <int, RecordId> > matchingTuples,
                                const vector<SelCond> cond, RecordFile rf,
                                int attr, const string table)
{
  RC     rc;
  int    key;
  string value;
  int    diff;
  RecordId rid;

  bool checkValueConditions = valueConditionExists(cond);
  sort(matchingTuples.begin(), matchingTuples.end(), sortTuplesByPid);

  int matchingTuplesCount = 0;
  for (int i = 0; i < matchingTuples.size(); ++i)
  {
    key = matchingTuples[i].first;
    rid = matchingTuples[i].second;

    bool readValue = false;
    if (checkValueConditions)
    {
      if ((rc = rf.read(rid, key, value)) < 0) {
        fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
        rf.close();
        return rc;
      }
      readValue = true;

      bool match = true;
      for (int j = 0; j < cond.size(); j++)
      {
        // compute the difference between the tuple value and the condition value
        if (cond[j].attr == VALUE_CONDITION)
        {
          diff = strcmp(value.c_str(), cond[j].value);
          // skip the tuple if any condition is not met
          switch (cond[j].comp) {
            case SelCond::EQ:
              if (diff != 0) match = false;
              break;
            case SelCond::NE:
              if (diff == 0) match = false;
              break;
            case SelCond::GT:
              if (diff <= 0) match = false;
              break;
            case SelCond::LT:
              if (diff >= 0) match = false;
              break;
            case SelCond::GE:
              if (diff < 0) match = false;
              break;
            case SelCond::LE:
              if (diff > 0) match = false;
              break;
          }
        }
        else if (cond[j].attr == KEY_CONDITION && cond[j].comp == SelCond::NE)
        {
          int searchKey = atoi(cond[j].value);
          if (key == searchKey) match = false;
        }
      }

      if (!match) continue;
    }

    // print the tuple
    switch (attr) {
      case 1:  // SELECT key
        fprintf(stdout, "%d\n", key);
        break;
      case 2:  // SELECT value
        rid = matchingTuples[i].second;
        if (!readValue)
        {
          if ((rc = rf.read(rid, key, value)) < 0) {
            fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
            rf.close();
            return rc;
          }
          readValue = true;
        }
        fprintf(stdout, "%s\n", value.c_str());
        break;
      case 3:  // SELECT *
        rid = matchingTuples[i].second;
        if (!readValue)
        {
          if ((rc = rf.read(rid, key, value)) < 0) {
            fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
            rf.close();
            return rc;
          }
          readValue = true;
        }
        fprintf(stdout, "%d '%s'\n", key, value.c_str());
        break;
      case 4:
        matchingTuplesCount += 1;
        break;
    }
  }

  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", matchingTuplesCount);
  }
  return 0;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile rf;   // RecordFile containing the table
  RecordId   rid;  // record cursor for table scanning

  BTreeIndex index;

  RC     rc;
  int    key;     
  string value;
  int    count;
  int    diff;

  // open the table file
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
   fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  if ((rc = index.open(table + ".idx", 'r')) ==  0)
  {
    // Optimization for selecting key when there are no WHERE conditions.
    vector<SelCond> conds = cond;
    if (conds.empty() && attr == 1)
    {
      SelCond noCondCond;
      noCondCond.attr = 1;
      noCondCond.comp = SelCond::GE;
      string s = "" + index.minKey;
      noCondCond.value = const_cast<char*>(s.c_str());
      conds.push_back(noCondCond);
    }

    if (keyConditionExists(conds))
    {
      vector< pair<int, RecordId> > matchingTuples;
      findMatchingTuples(index, conds, matchingTuples);
      checkValueConditionsAndPrint(matchingTuples, conds, rf, attr, table);
      return 0;
    }
    // else we can't use the index to optimize anything, so just use do a basic table scan.
  }
  // else index file does not exist or we can't open it, so just do a basic table scan.

  // scan the table file from the beginning
  rid.pid = rid.sid = 0;
  count = 0;
  while (rid < rf.endRid()) {
    // read the tuple
    if ((rc = rf.read(rid, key, value)) < 0) {
      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
      goto exit_select;
    }

    // check the conditions on the tuple
    for (unsigned i = 0; i < cond.size(); i++) {
      // compute the difference between the tuple value and the condition value
      switch (cond[i].attr) {
      case 1:
	diff = key - atoi(cond[i].value);
	break;
      case 2:
	diff = strcmp(value.c_str(), cond[i].value);
	break;
      }

      // skip the tuple if any condition is not met
      switch (cond[i].comp) {
      case SelCond::EQ:
	if (diff != 0) goto next_tuple;
	break;
      case SelCond::NE:
	if (diff == 0) goto next_tuple;
	break;
      case SelCond::GT:
	if (diff <= 0) goto next_tuple;
	break;
      case SelCond::LT:
	if (diff >= 0) goto next_tuple;
	break;
      case SelCond::GE:
	if (diff < 0) goto next_tuple;
	break;
      case SelCond::LE:
	if (diff > 0) goto next_tuple;
	break;
      }
    }

    // the condition is met for the tuple. 
    // increase matching tuple counter
    count++;

    // print the tuple 
    switch (attr) {
    case 1:  // SELECT key
      fprintf(stdout, "%d\n", key);
      break;
    case 2:  // SELECT value
      fprintf(stdout, "%s\n", value.c_str());
      break;
    case 3:  // SELECT *
      fprintf(stdout, "%d '%s'\n", key, value.c_str());
      break;
    }

    // move to the next tuple
    next_tuple:
    ++rid;
  }

  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", count);
  }
  rc = 0;

  // close the table file and return
  exit_select:
  rf.close();
  return rc;
}

bool valueConditionExists(const vector<SelCond> cond)
{
  for (int j = 0; j < cond.size(); j++)
  {
    if (cond[j].attr == VALUE_CONDITION || cond[j].comp == SelCond::NE)
    {
      return true;
    }
  }
  return false;
}

bool keyConditionExists(const vector<SelCond> cond)
{
  for (int j = 0; j < cond.size(); j++)
  {
    if (cond[j].attr == KEY_CONDITION && !(cond[j].comp == SelCond::NE))
    {
      return true;
    }
  }
  return false;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  /* your code here */
  int recordCount = 0;
  ifstream infile (loadfile.c_str(), ios::in);
  if (infile.is_open())
  {
    RecordFile rf;
    int error;
    
    if ((error = rf.open(table + ".tbl", 'w') == 0))
    {
      BTreeIndex tree;
      if (index) 
      {
          error = tree.open(table + ".idx", 'w');
          if (error) cout << "Error Code " << error << ": Could not open index file!" << endl;
      }
      string line;
      while (infile.good())
      {
        getline(infile, line);
        int key;
        string value;
        if (line.compare("") != 0)
        {
          SqlEngine::parseLoadLine(line, key, value);
          RecordId rid = rf.endRid();
          error = rf.append(key, value, rid);
          recordCount += 1;
                    
          if (!error && index)
          {
              tree.insert(key, rid);
          }
        }
      }
      infile.close();
      rf.close();
      if(index) tree.close();
    }
    else cout << "Error Code " << error << ": Could not open record file!";
  }
  else cout << "Error opening load file!" << endl;

  return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
